package com.wsjj.yjh;

import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;


public class DataStreamJsonSinkDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> source = env.fromElements("{\"id\": 111,\"name\": \"zhangsan\",\"age\": 14}");

        Properties properties = new Properties();
        properties.setProperty("format","json");
        properties.setProperty("strip_outer_array","true");

//
//         source.addSink(DorisSink.sink(
//                DorisReadOptions.builder().build(),
//                DorisExecutionOptions.builder()
//                        .setBatchSize(3)
//                        .setBatchIntervalMs(0L)
//                        .setMaxRetries(3)
//                        .setStreamLoadProp(properties)
//                        .build(),
//                DorisOptions.builder()
//                        .setFenodes("192.168.10.105:8030")
//                        .setTableIdentifier("test_doris.student_kafka")
//                        .setUsername("root")
//                        .setPassword("000000")
//                        .build()
//        ));

        source.addSink(DorisSink.sink(DorisOptions.builder()
                        .setFenodes("192.168.10.105:8030")
                        .setTableIdentifier("test_doris.student_kafka")
                        .setUsername("root")
                        .setPassword("000000")

                .build()));



        env.execute();

    }

}
